/**
 * @fileName: RocketMQConfigConsumerListen.java
 * @author: pjf
 * @date: 2019年10月17日 下午3:22:40
 */

package com.rgs.base.rocketmq.config;

import com.google.common.base.Strings;
import com.rgs.base.rocketmq.RocketMqConsumer;
import com.rgs.base.rocketmq.RocketMqListenDoWorkHandle;
import com.rgs.base.rocketmq.annotation.RocketMqConsumerListenAnnotation;
import lombok.extern.slf4j.Slf4j;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.annotation.Order;

import javax.annotation.Resource;
import java.util.Map;

/**
 * @author Ren Gaoshuai
 * @date 2022-03-08 16:50
 **/
@Configuration
@Order
@Slf4j
@ConditionalOnProperty(prefix = "wxhj.rocket",
        name = "enable", havingValue = "true")
public class RocketMqConfigConsumerListen implements ApplicationContextAware, InitializingBean {
    static final String APPLICATION_NAME = "${spring.application.name}";
    @Resource
    RocketMqSource rocketMqSource;
    @Value(APPLICATION_NAME)
    String applicationName;

    private ConfigurableApplicationContext applicationContext;

    public RocketMqConfigConsumerListen() {
        super();
    }

    @Override
    public void afterPropertiesSet() throws Exception {

        Map<String, Object> beans = this.applicationContext
                .getBeansWithAnnotation(RocketMqConsumerListenAnnotation.class);
        beans.forEach(this::registerContainer);

    }

    private void registerContainer(String beanName, Object bean) {
        Class<?> clazz = AopUtils.getTargetClass(bean);
        //
        if (!RocketMqListenDoWorkHandle.class.isAssignableFrom(bean.getClass())) {
            throw new IllegalStateException(
                    clazz + " is not instance of " + RocketMqListenDoWorkHandle.class.getName());
        }
        RocketMqListenDoWorkHandle rocketMqListenDoWorkHandle = (RocketMqListenDoWorkHandle) bean;

        RocketMqConsumerListenAnnotation rocketMqConsumerListenAnnotation = clazz
                .getAnnotation(RocketMqConsumerListenAnnotation.class);
        BeanDefinitionBuilder beanBuilder = BeanDefinitionBuilder
                .rootBeanDefinition(rocketMqConsumerListenAnnotation.rocketMQClass());
        if (!Strings.isNullOrEmpty(rocketMqConsumerListenAnnotation.rocketGroup())) {
            //
            String rocketGroupTemp = rocketMqConsumerListenAnnotation.rocketGroup();
            rocketGroupTemp = rocketGroupTemp.replace(APPLICATION_NAME, applicationName);
            //
            rocketMqSource.setRocketGroup(rocketGroupTemp);
        }
        // beanBuilder.addConstructorArgValue(rocketMqSource);

        //
        beanBuilder.addConstructorArgValue(rocketMqSource);
        // .addConstructorArgValue(rocketMqSource.getRocketGroup());

        String containerBeanName = String.format("%s_%s", RocketMqConsumer.class.getName(),
                rocketMqSource.getRocketGroup());

        DefaultListableBeanFactory beanFactory = (DefaultListableBeanFactory) applicationContext.getBeanFactory();
        beanFactory.registerBeanDefinition(containerBeanName, beanBuilder.getBeanDefinition());

        RocketMqConsumer rocketMqConsume = beanFactory.getBean(containerBeanName,
                rocketMqConsumerListenAnnotation.rocketMQClass());

        if (!rocketMqConsume.isStarted()) {
            String topicTemp = rocketMqConsumerListenAnnotation.topic();
            //动态topic兼容
            topicTemp = topicTemp.replace(APPLICATION_NAME, applicationName);
            try {
//                rocketMqConsume.setMessageModel(rocketMqConsumerListenAnnotation.messageModel());
//
//                rocketMqConsume.setTopicAndSubExpression(
//                        topicTemp,
//                        rocketMqConsumerListenAnnotation.subExpression());
                rocketMqConsume.setRocketMqConsumerAnnotation(rocketMqConsumerListenAnnotation);
                rocketMqConsume.setConsumerHandle(q -> {
                    rocketMqListenDoWorkHandle.dataHandle(q);
                });
                rocketMqConsume.start();
            } catch (Exception e) {
                log.error("started rocketMqConsumer failed. {}", rocketMqConsume, e);
                throw new RuntimeException(e);
            }
        }
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = (ConfigurableApplicationContext) applicationContext;
    }
}
